package com.moon.game.client;

import java.net.InetSocketAddress;
import java.util.concurrent.Executors;

import org.apache.log4j.Logger;
import org.jboss.netty.bootstrap.ConnectionlessBootstrap;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.Channels;
import org.jboss.netty.channel.FixedReceiveBufferSizePredictorFactory;
import org.jboss.netty.channel.socket.DatagramChannelFactory;
import org.jboss.netty.channel.socket.nio.NioDatagramChannelFactory;
import org.jboss.netty.handler.codec.frame.DelimiterBasedFrameDecoder;
import org.jboss.netty.handler.codec.frame.Delimiters;
import org.jboss.netty.handler.codec.protobuf.ProtobufDecoder;
import org.jboss.netty.handler.codec.protobuf.ProtobufEncoder;
import org.jboss.netty.handler.codec.protobuf.ProtobufVarint32FrameDecoder;
import org.jboss.netty.handler.codec.protobuf.ProtobufVarint32LengthFieldPrepender;
import org.jboss.netty.handler.codec.string.StringDecoder;
import org.jboss.netty.handler.codec.string.StringEncoder;

import com.moon.game.clazz.PBMessageProto.PBMessage;
import com.moon.game.command.RequestSession;
import com.moon.game.constants.GlobalConstants;
import com.moon.game.server.SessionPool;

public class UdpClient {
	private static final Logger log = Logger.getLogger(UdpClient.class);
	public static int CODEC_PROTOBUF = 1;
	public static int CODEC_STRING = 2;
	private Channel channel = null;
	private String ip;
	private int port;
	private int codec;
	private int sendBufferSize;
	private int recvBufferSize;

	/**
	 * 初始化连接参数
	 * 
	 * @param ip
	 * @param port
	 * @param bufferSize
	 * @param codec
	 */
	public UdpClient(String ip, int port, int codec, int sendBufferSize, int recvBufferSize) {
		this.ip = ip;
		this.port = port;
		this.codec = codec;
		this.sendBufferSize = sendBufferSize;
		this.recvBufferSize = recvBufferSize;
		connect();
	}

	/**
	 * 发送
	 * 
	 * @param rsession
	 * @param req
	 * @param reSendFlag
	 *            重发标记 0-不需要重发，做超时处理 1-需要重发 2-超时忽略
	 */
	public void send(RequestSession reqSession, PBMessage request, int reSendFlag) {
		// 返回
		if (reSendFlag == 0) {
			reqSession.setLast_req(null);
			reqSession.setLast_udpclient(this);
			reqSession.setReSendFlag(reSendFlag);
		} else if (reSendFlag == 1) {
			// 超时重发
			reqSession.setLast_req(request);
			reqSession.setLast_udpclient(this);
			reqSession.setReSendFlag(reSendFlag);
		} else if (reSendFlag == 2) {
			// 超时忽略,执行下个状态
			reqSession.setLast_req(null);
			reqSession.setLast_udpclient(null);
			reqSession.setReSendFlag(reSendFlag);
		}

		// 超时处理
		SessionPool.UniData uniData = new SessionPool.UniData(SessionPool.UniData.FROM_TIMER, reqSession.getSessionId(), reqSession.getEvent(), request.getCmd(), request.getStatus());
		SessionPool.inst.addTimeout(uniData);
		try {
			if (!channel.isConnected()) {
				connect();
			}
			// 请求包大小限制
			if (request.hasData() && request.toByteArray().length >= this.sendBufferSize) {
				log.error("send request data too larger, pid:" + request.getId() + "cmd:" + request.getCmd() + ",size:" + request.toByteArray().length);
			} else {
				channel.write(request);
			}
		} catch (Exception e) {
			log.error(e.getMessage(), e);
		}
	}

	/**
	 * protocol
	 * 
	 * @author sky
	 * 
	 */
	private static class ProtoBufPipelineFactory implements ChannelPipelineFactory {
		public ChannelPipeline getPipeline() throws Exception {
			// Create a default pipeline implementation.
			ChannelPipeline pipeline = Channels.pipeline();
			pipeline.addLast("frameDecoder", new ProtobufVarint32FrameDecoder());
			pipeline.addLast("protobufDecoder", new ProtobufDecoder(PBMessage.getDefaultInstance()));

			pipeline.addLast("frameEncoder", new ProtobufVarint32LengthFieldPrepender());
			pipeline.addLast("protobufEncoder", new ProtobufEncoder());
			pipeline.addLast("handler", new UdpClientHandler());
			return pipeline;
		}
	}

	/**
	 * 字符串
	 * 
	 * @author sky
	 * 
	 */
	private class StringPipelineFactory implements ChannelPipelineFactory {
		public ChannelPipeline getPipeline() throws Exception {
			// Create a default pipeline implementation.
			ChannelPipeline pipeline = Channels.pipeline();
			pipeline.addLast("framer", new DelimiterBasedFrameDecoder(recvBufferSize, Delimiters.lineDelimiter()));
			pipeline.addLast("decoder", new StringDecoder());
			pipeline.addLast("encoder", new StringEncoder());

			pipeline.addLast("handler", new UdpClientHandler());
			return pipeline;
		}
	}

	/**
	 * 创建连接
	 */
	private void connect() {
		log.debug("connect. ip:" + ip + ",port:" + port);
		try {
			if (channel != null) {
				channel.close();
			}

			DatagramChannelFactory factory = new NioDatagramChannelFactory(Executors.newCachedThreadPool());
			ConnectionlessBootstrap bootstrap = new ConnectionlessBootstrap(factory);
			bootstrap.setPipelineFactory(CODEC_PROTOBUF == codec ? new ProtoBufPipelineFactory() : new StringPipelineFactory());
			bootstrap.setOption(GlobalConstants.NETTY_RECEIVE_BUFFER_SIZE_PREDICTOR_FACTORY, new FixedReceiveBufferSizePredictorFactory(recvBufferSize));
			channel = bootstrap.connect(new InetSocketAddress(ip, port)).getChannel();
		} catch (Exception e) {
			log.error("connect error.+ip:" + ip + ",port:" + port);
		}
	}

	/**
	 * 无返回请求(protobuf)
	 * 
	 * @param req
	 */
	public void sendNoReturn(PBMessage request) {
		try {
			// 请求包大小限制
			if (request.hasData() && request.toByteArray().length >= this.sendBufferSize) {
				log.error("send request data too larger, pid:" + request.getId() + "cmd:" + request.getCmd() + ",size:" + request.toByteArray().length);
			} else {
				channel.write(request);
			}
		} catch (Exception e) {
			log.error("request error,ip=" + ip + ",port=" + port, e);
		}
	}

	/**
	 * 无返回请求(string)
	 * 
	 * @param msg
	 */
	public void sendNoReturn(String msg) {
		try {
			channel.write(msg + "\r\n");
		} catch (Exception e) {
			log.error("request error,ip=" + ip + ",port=" + port, e);
		}
	}

	public String getIp() {
		return ip;
	}

	public void setIp(String ip) {
		this.ip = ip;
	}

	public int getPort() {
		return port;
	}

	public void setPort(int port) {
		this.port = port;
	}

}
